package com.bw.utils;

import com.bw.bean.Shop;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Timestamp;
import java.util.Random;

public class ShopSource implements SourceFunction<Shop> {
    private boolean isRunning = true;
    private final Random random = new Random();
    private final String[] businessTypes = {"餐饮", "零售", "服装", "电子产品", "美妆", "家居"};
    private final String[] addresses = {"北京市朝阳区", "上海市静安区", "广州市天河区", "深圳市南山区", "杭州市西湖区", "南京市玄武区"};

    @Override
    public void run(SourceContext<Shop> ctx) throws Exception {
        long shopId = 1L;
        while (isRunning) {
            Shop shop = new Shop();
            shop.setShopId(shopId++);
            shop.setShopName(businessTypes[random.nextInt(businessTypes.length)] + "店" + shopId);
            shop.setAddress(addresses[random.nextInt(addresses.length)]);
            shop.setBusinessType(businessTypes[random.nextInt(businessTypes.length)]);
            shop.setOpenTime(new Timestamp(System.currentTimeMillis() - random.nextInt(365 * 24 * 3600 * 1000 * 3)));
            shop.setManager("经理" + shopId);

            ctx.collect(shop);

            // 控制数据生成速度
            Thread.sleep(200);

            // 生成10个店铺后停止
            if (shopId > 10) {
                break;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
